/*
 * SocketConnectionResolver.java    0.0.1    11/dic/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transparency.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.mentalsmash.tazio.commons.identifiers.NID;
import org.mentalsmash.tazio.net.cs.client.Connection;
import org.mentalsmash.tazio.net.cs.client.ConnectionStateExeption;
import org.mentalsmash.tazio.net.cs.client.RemoteServer;
import org.mentalsmash.tazio.net.cs.client.SocketConnection;
import org.mentalsmash.tazio.net.cs.client.SocketRemoteServer;
import org.mentalsmash.tazio.transparency.commons.messages.BeginConnectionMessage;
import org.mentalsmash.tazio.transparency.commons.messages.ConnectionEstablished;
import org.mentalsmash.tazio.transparency.commons.messages.DisconnectionMessage;
import org.mentalsmash.tazio.transparency.commons.messages.TazioMessage;
import org.mentalsmash.tazio.transparency.interfaces.DeliveryFailureListener;
import org.mentalsmash.tazio.transparency.interfaces.Dispatcher;
import org.mentalsmash.tazio.transparency.interfaces.MessageFactory;
import org.mentalsmash.tazio.transparency.interfaces.Resolver;
import org.mentalsmash.tazio.transparency.interfaces.TransparencyMonitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.internal.Preconditions;
import com.google.inject.name.Named;

/**
 * This is the base implementation for the {@link Resolver} interface, it uses
 * an {@link HashMap} to store the mappings between
 * {@link org.mentalsmash.tazio.commons.identifiers.TazioNodeID}s and the
 * physical addresses of Replica Manager of the system.
 * <p>
 * The connection is provided by using a {@link RemoteServer} and a
 * {@link Connection} object, which are binded to a {@link SocketRemoteServer}
 * and to a {@link SocketConnection} respectively.
 * </p>
 * <p>
 * If extensions have to created using different communication supports, this
 * {@link Resolver} implementation should substituted whit another one, which
 * would have to provide a different implementation to the {@link Connection}
 * interface, respecting its contract, but reflecting the behavior of the
 * underlying communication layer.
 * </p>
 * 
 * @version 0.0.1 11/dic/2009
 * @author Enrico Savioli <savioli.enrico@gmail.com>
 */
public class WeakReplicaManagerResolver implements Resolver,
	DeliveryFailureListener {

    private final Logger log = LoggerFactory
	    .getLogger(WeakReplicaManagerResolver.class);

    private final List<Association> associations;

    private final RemoteServer remoteServer;

    private final int MAX_CONNECTION_REPETITIONS;
    private final int DEFAULT_SERVER_LISTENING_PORT;
    private final int POOL_SIZE;
    private final int MAX_POOL_SIZE;
    private final long KEEP_ALIVE_TIME;

    private final MessageFactory messageFactory;
    private final TransparencyMonitor tester;
    private final LinkedBlockingQueue<Runnable> queue;
    private final ThreadPoolExecutor executor;

    // patch fields to make the BeginConnectionMessage work
    private int localListeningPort = -1;
    private InetAddress localListeningAddress = null;

    @Inject
    public WeakReplicaManagerResolver(
	    Dispatcher dispatcher, // used only for register sudden
	    // disconnection messages
	    TransparencyMonitor tester,
	    MessageFactory messageFactory,
	    @Named("max_connection_repetitions") int MAX_CONNECTION_REPETITIONS,
	    @Named("default_listening_port") int DEFAULT_LISTENING_PORT,
	    @Named("default_server_listening_port") int DEFAULT_SERVER_LISTENING_PORT,
	    @Named("pool_size") int POOL_SIZE,
	    @Named("max_pool_size") int MAX_POOL_SIZE,
	    @Named("keep_alive_time") int KEEP_ALIVE_TIME,
	    @Named("default_replica_manager") String DEFAULT_REPLICA_MANAGER) {

	log.debug("Initializing the resolver");

	this.associations = new ArrayList<Association>();

	this.MAX_CONNECTION_REPETITIONS = MAX_CONNECTION_REPETITIONS;
	this.DEFAULT_SERVER_LISTENING_PORT = DEFAULT_SERVER_LISTENING_PORT;
	this.POOL_SIZE = POOL_SIZE;
	this.MAX_POOL_SIZE = MAX_POOL_SIZE;
	this.KEEP_ALIVE_TIME = KEEP_ALIVE_TIME;

	this.messageFactory = messageFactory;
	this.tester = tester;

	queue = new LinkedBlockingQueue<Runnable>();
	executor = new ThreadPoolExecutor(POOL_SIZE, MAX_POOL_SIZE,
		KEEP_ALIVE_TIME, TimeUnit.SECONDS, queue);

	this.remoteServer = new SocketRemoteServer(executor, null,
		this.DEFAULT_SERVER_LISTENING_PORT);

	// Actually the DEFAULT_REPLICA_MANAGER property is a comma separated
	// list of values
	//FIXME: Test whether this works
	String[] replicaManagerAddresses = DEFAULT_REPLICA_MANAGER.split("\\s*,\\s*");
	configureAssociations(replicaManagerAddresses);

	// ask the dispatcher to notify for replica manager disconnections
	dispatcher.register(DisconnectionMessage.class,
		new DisconnectionMessageHandler(this));

	log.debug("Resolver initialized");

    }

    /**
     * @param replicaManagerAddresses
     * 
     */
    private void configureAssociations(String[] replicaManagerAddresses) {

	for (String add : replicaManagerAddresses) {
	    try {
		InetAddress address = InetAddress.getByName(add);

		// FIXME: Is it ok that a weak replica manger GENERATES the node
		// id of a ReplicaManager??
		NID id = NID.generateNID();

		associations.add(new Association(id, address));

	    } catch (UnknownHostException e) {
		String msg = "Error while initializing the Association list.";
		log.error(msg, e);
	    }
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Resolver#registerMapping(org.mentalsmash
     * .tazio.core.identifiers.TazioNodeID, java.net.InetAddress)
     */
    @Override
    public void registerMapping(NID nid, InetAddress address) {

	tester.Resolver_registerMapping();

	Association alreadyContained = null;

	synchronized (associations) {

	    log.debug("Adding a new Association for: " + nid + "@"
		    + address.getCanonicalHostName());

	    for (int i = 0; i < associations.size(); i++) {

		alreadyContained = associations.get(i);

		// if an entry for the same NID is already contained it is
		// removed
		// only if they are not equal, in this case nothing happens.
		if (alreadyContained.getID().equals(nid)
			&& !alreadyContained.getAddress().equals(address)) {
		    log
			    .debug("An existing association already existed. Removing it.");
		    associations.remove(i);
		    break;
		}

	    }

	    associations.add(new Association(nid, address));
	    log.debug("Association added.");
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Resolver#removeMapping(org.mentalsmash
     * .tazio.core.identifiers.TazioNodeID, java.net.InetAddress)
     */
    @Override
    public void removeMapping(NID nid) {

	tester.Resolver_removeMapping();

	synchronized (associations) {

	    log.debug("Removing the binding for " + nid);
	    Association toRemove = null;

	    for (int i = 0; i < associations.size(); i++) {

		toRemove = associations.get(i);
		if (toRemove.getID().equals(nid)) {

		    log.debug("Removed the binding for " + nid);
		    associations.remove(i);
		    return;

		}

	    }

	    log.debug("Binding for " + nid + " not found!");
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Resolver#resolve(org.mentalsmash.tazio
     * .core.identifiers.TazioNodeID)
     */
    @Override
    public Connection resolve(NID nid) throws TransparentDeliveryException {

	// System.out.println("Resolving an address");

	tester.Resolver_resolve();

	synchronized (associations) {

	    Association toRetrieve = null;
	    InetAddress toConnect = null;
	    InetAddress currentAddress = remoteServer.getAddress();
	    Connection connection = null;

	    toConnect = getAddressFromNodeID(nid);

	    // checks if the remote server is already connected to the desired
	    // address.
	    if (!currentAddress.equals(toConnect)) {

		// if the remote server is connected it disconnects it.
		if (remoteServer.isConnected()) {
		    remoteServer.disconnect();

		    // updates the order of the servers, the one it is connected
		    // to will
		    // always be the first entry in the list

		    // FIXME: Doesn't this duplicate the entry?
		    associations.add(0, toRetrieve);
		}

		// starts the connection protocol.
		connection = connect(nid, toConnect);

		if (connection == null) {
		    // this exception is thrown if the connection to the server
		    // failed.
		    throw new TransparentDeliveryException();
		}

	    } else {

		// returns the current connection if the remote server is
		// already connected
		// to the desired Replica Manager
		connection = remoteServer.getConnection();
	    }

	    return connection;

	}

    }

    /**
     * This method has the task of connecting the remote server object to the
     * Replica Manager residing at the provided address.
     * <p>
     * To do that tries to change the address associated with the remote server
     * object, expecting that the object had previously been disconnected,
     * otherwise an exception would be thrown and caught, leading to the
     * disconnection of the remote server object and the continuing of the
     * connection protocol.
     * </p>
     * <p>
     * After the remote server's address has been changed, this method tries to
     * connect it to the new remote Replica Manager, using the
     * <code>connect()</code> method. If any problem occurs an exception will be
     * thrown and caught: nothing will be done to handle this kind of exception,
     * except from logging the error and continuing in the loop.
     * </p>
     * <p>
     * The connection protocol is repeated a number of
     * <code>MAX_CONNECTION_REPETITIONS
	 * </code> times or until it successfully completes, without the throwing of
     * any exception, and will return a {@link Connection} object to the caller,
     * which will be the correct connection to the Replica Manager if no problem
     * occurred, or a <code>null
	 * </code> value if the connection protocol failed
     * <code>MAX_CONNECTION_REPETITIONS</code> times.
     * </p>
     * 
     * @param toConnect
     *            the {@link InetAddress} of the Replica Manager to which the
     *            {@link RemoteServer} has to be connected.
     * @return the {@link Connection} over which communicate with the Replica
     *         Manager or a <code>null</code> value if the connection protocol
     *         failed.
     */
    private Connection connect(NID nid, InetAddress toConnect) {

	Connection connection = null;
	BeginConnectionMessage conMessg = new BeginConnectionMessage(
		localListeningPort, localListeningAddress);
	TazioMessage connectionMessage = messageFactory
		.encapsulateMessage(conMessg);
	TazioMessage answer = null;

	for (int i = 0; i < MAX_CONNECTION_REPETITIONS; i++) {

	    try {

		// updating the address to which the RemoteServer will connect.
		remoteServer.setAddress(toConnect);

	    } catch (ConnectionStateExeption e) {
		String msg = "Trying to change the address of the remote server while "
			+ "it is still connected, disconnecting it and retrying to set its "
			+ "address";
		log.error(msg);
		e.printStackTrace();
		remoteServer.disconnect();
		continue;

	    }
	    try {

		// trying to connect the RemoteServer to the new address.
		connection = remoteServer.connect();

	    } catch (IOException e) {
		String msg = "Error while connecting to the remote server, "
			+ toConnect.getAddress() + " attempt number: " + i;
		log.error(msg);
		e.printStackTrace();

		// after MAX_CONNECTION_REPETITIOMS time the non-working Replica
		// Manager
		// is placed at the end of the list.
		if (i == (MAX_CONNECTION_REPETITIONS - 1)) {

		    Association toRemove = getAssociationFromNodeID(nid);
		    associations.remove(toRemove);
		    associations.add(toRemove);

		}

		continue;
	    }

	    break;
	}

	for (int i = 0; i < MAX_CONNECTION_REPETITIONS; i++) {

	    try {

		answer = (TazioMessage) connection
			.sendRequest(connectionMessage);

	    } catch (ConnectionStateExeption e) {

		String msg = "Error while connecting to a ReplicaManager. The connection "
			+ "is in an invalid state.";
		log.error(msg);
		e.printStackTrace();

	    } catch (IOException e) {

		String msg = "Input/Output error occurred while connecting to the "
			+ "ReplicaManager.";
		log.error(msg);
		e.printStackTrace();

	    }

	    if (((TazioMessage) answer).getPayload() instanceof ConnectionEstablished) {

		log.debug("Connection successfully established!");
		break;

	    }

	    // After MAX_CONNECTION_REPETITIONS times that the ReplicaManager
	    // does not
	    // accept the connection, the RM is put at the end of the list and
	    // the connection
	    // protocol ends for that RM.
	    if (i == (MAX_CONNECTION_REPETITIONS - 1)) {

		Association toRemove = getAssociationFromNodeID(nid);
		associations.remove(toRemove);
		associations.add(toRemove);

	    }

	}

	// Returns the connection to the ReplicaManager, if problems occurred
	// during the
	// connection protocol this object is null.
	return connection;

    }

    public int getPoolSize() {

	return this.POOL_SIZE;

    }

    public int getMaxPoolSize() {

	return this.MAX_POOL_SIZE;

    }

    public long getKeepAliveTime() {

	return this.KEEP_ALIVE_TIME;

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Resolver#getAddress(org.mentalsmash
     * .tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public InetAddress getAddress(NID nid) {

	tester.Resolver_getAddress();

	synchronized (associations) {

	    log.debug("Retrieving the address for " + nid);

	    Association toRetrieve;

	    for (int i = 0; i < associations.size(); i++) {

		toRetrieve = associations.get(i);

		if (toRetrieve.getID().equals(nid)) {

		    log.debug("Retrieved the address for: " + nid);
		    return toRetrieve.getAddress();

		}

	    }

	    log.debug("Address for " + nid + " not found!");
	    return null;
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.Resolver#getCurrentServer()
     */
    @Override
    public List<NID> getNIDList() {

	tester.Resolver_getNIDList();

	List<NID> toReturn = new ArrayList<NID>();

	for (int i = 0; i < associations.size(); i++) {

	    toReturn.add(associations.get(i).getID());

	}

	return toReturn;
    }

    private Association getAssociationFromNodeID(NID nid) {

	Association toRetrieve;

	for (int i = 0; i < associations.size(); i++) {

	    toRetrieve = associations.get(i);

	    if (toRetrieve.getID().equals(nid)) {

		log.debug("Retrieved the address for: " + nid);
		return toRetrieve;

	    }

	}

	log.debug("No association for " + nid + " found!");
	return null;
    }

    private InetAddress getAddressFromNodeID(NID nid) {

	Association toRetrieve;

	for (int i = 0; i < associations.size(); i++) {

	    toRetrieve = associations.get(i);

	    if (toRetrieve.getID().equals(nid)) {

		log.debug("Retrieved the address for: " + nid);
		return toRetrieve.getAddress();

	    }

	}
	log.debug("No Association for " + nid + " found!");
	return null;
    }

    private class Association {

	private final NID id;
	private final InetAddress address;

	public Association(NID id, InetAddress address) {

	    this.id = id;
	    this.address = address;

	}

	public NID getID() {

	    return this.id;

	}

	public InetAddress getAddress() {

	    return this.address;

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.interfaces.DeliveryFailureListener
     * #failedDelivery(org.mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public void failedDelivery(NID nid) {

	synchronized (associations) {

	    Association association = getAssociationFromNodeID(nid);

	    // Remove and re-add the association, it will be placed at the end
	    // of the list.
	    associations.remove(association);
	    associations.add(association);

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transparency.interfaces.Resolver#
     * setLocalListeningAddress(java.net.InetAddress, int)
     */
    @Override
    public synchronized void setLocalListeningAddress(InetAddress address,
	    int port) {
	Preconditions.checkNotNull(address);
	Preconditions.checkArgument(port > 0 && port < 65536);

	this.localListeningAddress = address;
	this.localListeningPort = port;

    }
}
